package com.doitedu.mr.day04.join;

import com.doitedu.mr.day04.bean.MovieWriable;
import com.doitedu.mr.day04.movie.TopN;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @Date 2021/12/3
 * @Created by HANGGE
 * @Description TODO
 */
public class Join {
    /**
     * 输出的是  K:uid  V:行内容
     */
    static  class JoinMapper extends Mapper<LongWritable , Text , Text, Text> {
        String fileName = null;

        // 当前task只执行一次
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            FileSplit inputSplit = (FileSplit) context.getInputSplit();
            fileName = inputSplit.getPath().getName();
        }

        Text k = new Text();
        Text v = new Text();

        // 每行执行一次 获取文件名
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String uid = null;
            // 处理数据   不确定处理的文件的内容  , 根据文件名
            if (fileName.startsWith("user")) { // 处理的是用户数据
                String[] split = line.split(",");
                uid = split[0];
            } else {// 订单数据
                String[] split = line.split("\\s+");
                uid = split[1];
            }
            k.set(uid);
            v.set(line);
            context.write(k, v);
        }


        static class JoinReducer extends Reducer<Text, Text, Text, NullWritable> {
            Text k = new Text();

            @Override
            protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
                String user = null;
                List<String> orders = new ArrayList<>();
                //遍历所有的内容
                for (Text value : values) {
                    String content = value.toString();
                    // 判断内容中包含 , 用户信息
                    if (content.contains(",")) {
                        // content用户数据
                        user = content;
                    } else { // 订单   将订单信息存储在集合中
                        orders.add(content);
                    }
                }
                // 遍历订单  拼接用户   left join
                for (String order : orders) {
                    // 拼接数据
                    String join = order.split("\\s+")[0] + "," + user;
                    // 设置数据
                    k.set(join);
                    // 将数据写出
                    context.write(k, NullWritable.get());
                }
            }
        }

        public static void main(String[] args) throws Exception {
            // Job  统计文件中单词出现的次数
            Configuration conf = new Configuration();
            // 1  创建 JOB
            Job job = Job.getInstance(conf, "join");
            // 1) Mapper 类
            job.setMapperClass(JoinMapper.class);
            // 2) Reducer类
            job.setReducerClass(JoinReducer.class);
            // 3)  设置map输出的数据类型
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(Text.class);
            // 4)  设置reduce输出的数据类型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(NullWritable.class);
            // 5) 设置输入的数据路径
            FileInputFormat.setInputPaths(job, new Path("E:\\mrdata\\join\\input"));
            // 6) 设置输出的结果保存路径
            FileOutputFormat.setOutputPath(job, new Path("E:\\mrdata\\join\\res"));
            // 2 提交工作
            // 等待执行完成 , 打印执行过
            job.waitForCompletion(true);
        }
    }
}
